package com.zhike.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.zhike.bean.EnergyChargeBean;
import com.zhike.function.BeanToJsonFunction;
import com.zhike.function.DimAsyncFunction;
import com.zhike.utils.DateFormatUtil;
import com.zhike.utils.DorisUtils;
import com.zhike.utils.KafkaUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.concurrent.TimeUnit;

/**
 * @Author: zhike
 * @CreateTime: 2024/2/27 17:00
 * @Description: 计算各车的次充电电量，充电时长，充电次数
 */
public class DwsEnergyChargeWindow {
    public static void main(String[] args) {
        //1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.enableCheckpointing(10000, CheckpointingMode.EXACTLY_ONCE);
//        env.getCheckpointConfig().setCheckpointTimeout(60*1000);
//        env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
//        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3,5000L));
//
//        //设置状态后端
//        env.setStateBackend(new HashMapStateBackend());
//        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop101:8020/gmall/DimApp/");

        //2.读取car_parking_charging_log日志数据创建流并转换为EnergyChargeBean对象
        SingleOutputStreamOperator<EnergyChargeBean> chargeDS = env.fromSource(KafkaUtils.getKafkaSource("car_parking_charging_log", "dws_energy_charge_window"),
                        WatermarkStrategy.noWatermarks()
                        , "car_parking_charging_log")
                .map(JSON::parseObject)
                .map(new MapFunction<JSONObject, EnergyChargeBean>() {
                    @Override
                    public EnergyChargeBean map(JSONObject value) throws Exception {
                        //获取vin
                        String vin = value.getString("vin");
                        //获取时间戳
                        Long ts = value.getLong("timestamp");
                        //获取当前电流
                        Integer electricCurrent = value.getInteger("electric_current");
                        //获取当前电量
                        Integer soc = value.getInteger("soc");
                        return EnergyChargeBean.builder()
                                .vin(vin)
                                .maxSoc(soc)
                                .minSoc(soc)
                                .chargeElectricCount(soc > 0 ? 1 : 0)
                                .chargeElectric(electricCurrent)
                                .maxTs(ts)
                                .minTs(ts)
                                .ts(ts)
                                .build();
                    }
                });
        
        //3.设置水位线
        SingleOutputStreamOperator<EnergyChargeBean> wMDS = chargeDS.assignTimestampsAndWatermarks(WatermarkStrategy
                .<EnergyChargeBean>forBoundedOutOfOrderness(Duration.ofSeconds(30L))
                .withTimestampAssigner(new SerializableTimestampAssigner<EnergyChargeBean>() {
                    @Override
                    public long extractTimestamp(EnergyChargeBean element, long recordTimestamp) {
                        return element.getTs();
                    }
                }));

        //4.根据vin分组
//        KeyedStream<EnergyChargeBean, String> keyByDS = wMDS.keyBy(
//                (KeySelector<EnergyChargeBean, String>) EnergyChargeBean::getVin);
        KeyedStream<EnergyChargeBean, String> keyByDS = wMDS.keyBy(new KeySelector<EnergyChargeBean, String>() {
            @Override
            public String getKey(EnergyChargeBean value) throws Exception {
                return value.getVin();
            }
        });

        //5.根据状态编程计算充电id
        SingleOutputStreamOperator<EnergyChargeBean> addChargeIdDS = keyByDS.process(new KeyedProcessFunction<String, EnergyChargeBean, EnergyChargeBean>() {
            private ValueState<Long> tsState;
            private ValueState<Long> chargeIdState;
            @Override
            public void open(Configuration parameters) throws Exception {
                tsState = getRuntimeContext().getState(new ValueStateDescriptor<Long>("ts_state", Long.class));
                chargeIdState = getRuntimeContext().getState(new ValueStateDescriptor<>("charge_id_state", Long.class));

            }

            @Override
            public void processElement(EnergyChargeBean value, KeyedProcessFunction<String, EnergyChargeBean, EnergyChargeBean>.Context ctx, Collector<EnergyChargeBean> out) throws Exception {
                //获取上一次充电id
                Long chargeID = chargeIdState.value();
                if (chargeID == null) {
                    chargeID = 1L;
                }
//                System.out.println("chargeID:" + chargeID);
                //获取上条日志时间戳
                Long tsPre = tsState.value();
                //获取当前日志时间戳
                Long ts = value.getTs();
                if (tsPre!=null){
                    if (ts<tsPre){
                        System.out.println("超时了ts = " + ts+" tsPre = " + tsPre+"tspre-ts = "+(tsPre-ts));
                        System.out.println(value);
                    }
                }
                if (tsPre == null) {
                    tsState.update(ts);
                    value.setChargeId(chargeID);
                } else if(ts-tsPre>60000) {
                    chargeIdState.update(chargeID + 1);
                    tsState.update(ts);
                    value.setChargeId(chargeID + 1);
                }else {
                    tsState.update(ts);
                    value.setChargeId(chargeID);
                }
                out.collect(value);
            }
        });


        //6.按照vin，充电id分组
        KeyedStream<EnergyChargeBean, Tuple2<String, Long>> keyDS = addChargeIdDS.keyBy(
                new KeySelector<EnergyChargeBean, Tuple2<String, Long>>() {
                    @Override
                    public Tuple2<String, Long> getKey(EnergyChargeBean value) throws Exception {
                        return Tuple2.of(value.getVin(), value.getChargeId());
                    }
                }
        );

        //7.开窗，聚合
        SingleOutputStreamOperator<EnergyChargeBean> reduceDS = keyDS.window(TumblingEventTimeWindows.of(Time.seconds(60L)))
                .reduce(new ReduceFunction<EnergyChargeBean>() {
                    @Override
                    public EnergyChargeBean reduce(EnergyChargeBean value1, EnergyChargeBean value2) throws Exception {
                        value1.setMaxTs(Math.max(value1.getMaxTs(), value2.getMaxTs()));
                        value1.setMinTs(Math.min(value1.getMinTs(), value2.getMinTs()));
                        value1.setMaxSoc(Math.max(value1.getMaxSoc(), value2.getMaxSoc()));
                        value1.setMinSoc(Math.min(value1.getMinSoc(), value2.getMinSoc()));
                        value1.setChargeElectric(value1.getChargeElectric() + value2.getChargeElectric());
                        value1.setChargeElectricCount(value1.getChargeElectricCount() + value2.getChargeElectricCount());
                        return value1;
                    }
                }, new WindowFunction<EnergyChargeBean, EnergyChargeBean, Tuple2<String, Long>, TimeWindow>() {
                    @Override
                    public void apply(Tuple2<String, Long> stringLongTuple2, TimeWindow window, Iterable<EnergyChargeBean> input, Collector<EnergyChargeBean> out) throws Exception {
                        EnergyChargeBean bean = input.iterator().next();
                        bean.setStt(DateFormatUtil.toYmdHms(window.getStart()));
                        bean.setEdt(DateFormatUtil.toYmdHms(window.getEnd()));
                        bean.setWindowDate(DateFormatUtil.toDate(window.getStart()));
                        bean.setTs(System.currentTimeMillis());
                        out.collect(bean);
                    }
                });

        //8.关联维表补充车型，品牌
        SingleOutputStreamOperator<EnergyChargeBean> rishDS = AsyncDataStream.unorderedWait(reduceDS, new DimAsyncFunction<EnergyChargeBean>("DIM_CAR_INFO") {
            @Override
            public String getKey(EnergyChargeBean input) {
                return input.getVin();
            }

            @Override
            public void join(EnergyChargeBean input, JSONObject dimInfo) {
                input.setCarType(dimInfo.getString("TYPE"));
                input.setBrand(dimInfo.getString("TRADEMARK"));
            }
        }, 100L, TimeUnit.SECONDS);

        //9.将结果写入doris
//        rishDS.print();
        SingleOutputStreamOperator<String> resultDS = rishDS.map(BeanToJsonFunction::getString);
//        resultDS.print();
        resultDS.sinkTo(DorisUtils.getDorisSink("dws_energy_charge_window"));

        //10.启动任务
        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}


